package com.imooc.springcloud.biz;

import com.imooc.springcloud.topic.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;

import java.util.concurrent.atomic.AtomicInteger;


/**
 * @describte:消费者
 * @program:spring-cloud-demo
 * @author:zhengdp
 * @create:2021-10-09 16:22:48
 **/
@Slf4j
// 将信道 绑定到 broker服务器
// 绑定之后 会在Rabbitmq中创建一个对应的Queue
@EnableBinding(
        value = {
                Sink.class,
                MyTopic.class,
                GroupTopic.class,
                DelayedTopic.class,
                ErrorTopic.class,
                RequeueTopic.class,
                DlqTopic.class,
                FallbackTopic.class
        }
)
public class StreamConsumer {

    private AtomicInteger count = new AtomicInteger(1);

    // 绑定信道
    @StreamListener(Sink.INPUT)
    public void consumer(Object payload){
        log.info("Message consumer successfully , payload={}",payload);
    }

    @StreamListener(MyTopic.INPUT)
    public void consumerMyMessage(Object payload){
        log.info("My Message consumer successfully , payload={}",payload);
    }

    @StreamListener(GroupTopic.INPUT)
    public void consumerGroupMessage(Object payload){
        log.info("Group Message consumer successfully , payload={}",payload);
    }

    @StreamListener(DelayedTopic.INPUT)
    public void consumeDelayedMessage(MessageBean bean){
        log.info("Delayed Message consumer successfully , payload={}",bean.getPayload());
    }

    // 异常重试（单机版）
    @StreamListener(ErrorTopic.INPUT)
    public void consumerErrorMessage(MessageBean bean){
        log.info("Are you ok ?");
        if(count.incrementAndGet() % 3 ==0){
            log.info("Fine , thank you. And you?");
            count.set(0);
        }else{
            log.info("What's your problem?");
            throw new RuntimeException("I'm not OK");
        }
        log.info("Delayed Message consumer successfully , payload={}",bean.getPayload());
    }

    // 异常重试 Re-queue
    @StreamListener(RequeueTopic.INPUT)
    public void requeueErrorMessage(MessageBean bean){
        log.info("Are you ok ?");
        try{
            Thread.sleep(3000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        throw new RuntimeException("I'm not OK");
    }

    // 死信队列
    @StreamListener(DlqTopic.INPUT)
    public void consumerDlqMessage(MessageBean bean){
        log.info("Dlq - Are you ok ?");
        if(count.incrementAndGet() % 3 ==0){
            log.info("Dlq - Fine , thank you. And you?");
        }else{
            log.info("Dlq - What's your problem?");
            throw new RuntimeException("I'm not OK");
        }
    }

    // Fallback + 升级版本
    @StreamListener(FallbackTopic.INPUT)
    public void goodbyeBadGuy(MessageBean bean,
                              @Header("version")String version){
        log.info("Fallback - Are you ok ?");
        if("1.0".equalsIgnoreCase(version)){
            log.info("Fallback - Fine , thank you. And you?");
        }else if("2.0".equalsIgnoreCase(version)){
            log.info("unsupported version");
            throw new RuntimeException("I'm not OK");
        }else{
            log.info("Fallback - version = {}",version);
        }
    }

    @ServiceActivator(inputChannel = "fallback-topic.fallback-group.errors")
    public void fallback(Message<?> message){
        log.info("fallback entered");
    }
}
